package com.gugee.ins.data.task.monitor.service;

import com.gugee.common.task.dto.TaskParam;
import com.gugee.ins.common.model.kafka.KafKaConsumerOffset;
import com.gugee.ins.data.common.constant.EventStatusEnum;
import com.gugee.ins.data.common.constant.EventTypeEnum;
import com.gugee.ins.data.common.msg.EventMsg;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.admin.*;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.*;

/**
 * ins 品牌日打点
 * @author huangxin
 */
@Component
@Slf4j
public class InsKafKaLagMonitorSvc extends AbstractMonitorProducerHandler {

    /**
     *
     */
    @Autowired
    @Qualifier("kafkaTemplateMonitor")
    private KafkaTemplate<Long,EventMsg> kafkaTemplateMonitor;

    /**
     * 执行
     */
    @Override
    public void handle(TaskParam taskParam) {
        // 获取连接实例
        try (AdminClient client = AdminClient.create(getAdminClientProperties())) {
            listConsumerOffset(client);
        }catch (Exception e){
            log.info("kafka监控队列积压任务失败 error, msg :{}", e.getMessage());
            XxlJobLogger.log("kafka监控队列积压任务失败 error, msg :{}", e.getMessage());
        }
    }

    /**
     * 获取配置
     * @return
     */
    private Properties getAdminClientProperties(){
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, taskConfig.getKafkaAdminClientServers());
        properties.put(ProducerConfig.CONNECTIONS_MAX_IDLE_MS_CONFIG, taskConfig.getKafkaAdminClientConnectionsIdle());
        properties.put(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG, taskConfig.getKafkaAdminClientRequestTimeout());
        return properties;
    }

    /**
     * 获取消费组的
     * @param client
     */
    public void listConsumerOffset(AdminClient client){
        // 当前时间
        long startTime = System.currentTimeMillis();
        // 获取所有消费组的信息
        ListConsumerGroupsResult listConsumerGroupsResult = client.listConsumerGroups();
        try {
            // 获取所有组的 groupId 集合
            Collection<ConsumerGroupListing> consumerGroupListings = listConsumerGroupsResult.all().get();
            // 用来存储group 和 TopicPartition 的关系
            Map<String, List<TopicPartition>> group4TopicPartition = new HashMap<>(consumerGroupListings.size());
            // 用来存储所有的 TopicPartition ，作为参数获取caption 的logSize
            Map<TopicPartition, OffsetSpec> topicPartitionOffsets = new HashMap<>(consumerGroupListings.size());
            // 用来存储各个 group-topic-Partition 的Offsets
            Map<String, Long> gtpOffsets = new HashMap<>(consumerGroupListings.size());

            // 获取各个 group 下，各个 TopicPartition 的 offset值
            consumerGroupListings.forEach(group -> getConsumerOffset(group.groupId(),client,topicPartitionOffsets,gtpOffsets,group4TopicPartition));

            // 用来存储结果
            Map<String, KafKaConsumerOffset> kco = new HashMap<>(topicPartitionOffsets.size());

            // 获取各个 TopicPartition 的logSize值
            ListOffsetsResult listOffsetsResult = client.listOffsets(topicPartitionOffsets);
            Map<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> lr = listOffsetsResult.all().get();
            Map<String,Long> logSizes = new HashMap<>(lr.size());
            Iterator<Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo>> lrIterator = lr.entrySet().iterator();
            while (lrIterator.hasNext()){
                Map.Entry<TopicPartition, ListOffsetsResult.ListOffsetsResultInfo> entry = lrIterator.next();
                // 结果集中设置logSize
                logSizes.put(entry.getKey().toString(),entry.getValue().offset());
            }
            // 超过阈值的队列
            Set<String> groupTopic = new HashSet<>(kco.size());

            // 汇总结果，计算lag
            Iterator<Map.Entry<String,List<TopicPartition>>> monitor = group4TopicPartition.entrySet().iterator();
            while(monitor.hasNext()){
                Map.Entry<String,List<TopicPartition>> entry = monitor.next();
                entry.getValue().forEach(topicPartition -> {
                    String key = entry.getKey() + "_" + topicPartition.topic();
                    // 获取 group topic 的信息
                    KafKaConsumerOffset kafKaConsumerOffsetDto = kco.get(key);
                    if(null == kafKaConsumerOffsetDto){
                        kafKaConsumerOffsetDto = new KafKaConsumerOffset();
                        kafKaConsumerOffsetDto.setGroupId(entry.getKey());
                        kafKaConsumerOffsetDto.setTopicPartition(topicPartition.topic());
                    }
                    kafKaConsumerOffsetDto.setOffset(kafKaConsumerOffsetDto.getOffset() + gtpOffsets.get(entry.getKey() + "_" + topicPartition.toString()));
                    kafKaConsumerOffsetDto.setLogSize(kafKaConsumerOffsetDto.getLogSize() + logSizes.get(topicPartition.toString()));
                    kafKaConsumerOffsetDto.setLag(kafKaConsumerOffsetDto.getLogSize() - kafKaConsumerOffsetDto.getOffset());
                    kco.put(key,kafKaConsumerOffsetDto);
                    // 判断是否超过阈值
                    if(kafKaConsumerOffsetDto.getLag() >= taskConfig.getInsMonitorKafkaLagCount()){
                        groupTopic.add(key);
                    }
                });
            }

            // 发送异常消息
            if(!groupTopic.isEmpty()){
                groupTopic.forEach(gt -> sendMsg(kco.get(gt),startTime));
            }
        } catch (Exception e) {
            log.info("监控 kafka任务 获取 kafka 队列信息 error, msg :{}", e.getMessage());
            XxlJobLogger.log("监控 kafka任务 获取 kafka 队列信息 error, msg :{}", e.getMessage());
        }
    }

    /**
     * 获取消费组消费各个topic的偏移量
     * @param groupId
     * @param client
     * @param topicPartitionOffsets
     * @param gtpOffsets
     * @param group4TopicPartition
     */
    public void getConsumerOffset(String groupId,AdminClient client,
                                         Map<TopicPartition, OffsetSpec> topicPartitionOffsets,
                                         Map<String, Long> gtpOffsets, Map<String,List<TopicPartition>> group4TopicPartition){
        // 获取各个 TopicPartition 的 offset值
        ListConsumerGroupOffsetsResult listConsumerGroupOffsetsResult = client.listConsumerGroupOffsets(groupId);
        try {
            Map<TopicPartition, OffsetAndMetadata> sr = listConsumerGroupOffsetsResult.partitionsToOffsetAndMetadata().get();
            List<TopicPartition> topicPartitionString = new ArrayList<>(sr.size());
            Iterator<Map.Entry<TopicPartition, OffsetAndMetadata>> iterator = sr.entrySet().iterator();
            while (iterator.hasNext()){
                // 保存offset信息
                Map.Entry<TopicPartition, OffsetAndMetadata> entry = iterator.next();
                String gt = groupId + "_" + entry.getKey().toString();
                gtpOffsets.put(gt,entry.getValue().offset());

                // topicPartitionOffsets 中添加topicPartition，用来查询logSize的参数
                topicPartitionOffsets.put(entry.getKey(),new OffsetSpec());
                topicPartitionString.add(entry.getKey());
            }

            // 添加 group 和 TopicPartition的关系信息
            group4TopicPartition.put(groupId,topicPartitionString);
        } catch (Exception e) {
            log.info("监控 kafka任务 获取 kafka 消费者offset信息 error, groupId:{} , msg :{}",groupId, e.getMessage());
            XxlJobLogger.log("监控 kafka任务 获取 kafka 消费者offset信息 error, groupId:{} , msg :{}",groupId, e.getMessage());
        }
    }

    /**
     * 发送消息
     * @param kafKaConsumerOffset
     * @param startTime
     */
    public void sendMsg(KafKaConsumerOffset kafKaConsumerOffset,long startTime){
        if(kafKaConsumerOffset != null && kafKaConsumerOffset.getLag() >= taskConfig.getInsMonitorKafkaLagCount()){
            long now = System.currentTimeMillis();
            StringBuilder content = new StringBuilder("groupId: ")
                    .append(kafKaConsumerOffset.getGroupId())
                    .append(" ,topic: ")
                    .append(kafKaConsumerOffset.getTopicPartition())
                    .append(" ,logSize: ")
                    .append(kafKaConsumerOffset.getLogSize())
                    .append(" ,offset: ")
                    .append(kafKaConsumerOffset.getOffset())
                    .append(" ,lag: ")
                    .append(kafKaConsumerOffset.getLag());
            EventMsg eventMsg = new EventMsg();
            eventMsg.setId(now);
            eventMsg.setOwner(kafKaConsumerOffset.getGroupId());
            eventMsg.setPosition(kafKaConsumerOffset.getTopicPartition());
            eventMsg.setEventStatus(EventStatusEnum.KAFKA_LAG.getCode());
            eventMsg.setEventType(EventTypeEnum.KAFKA_EVENT);
            eventMsg.setContent(content.toString());
            eventMsg.setSpendMills(now - startTime);
            kafkaTemplateMonitor.send(taskConfig.getInsMonitorOnlineTopic(),eventMsg);
        }
    }
}
